package org.example.realtime.traffic.dwd.log.function;

/**
 * @Title: FlowReportFunction
 * @Author Lianzy
 * @Package org.example.realtime.traffic.dwd.log.function
 * @Date 2025/5/29 18:09
 * @description
 */
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.realtime.jtp.common.utils.DateTimeUtil;

public class FlowReportFunction implements WindowFunction<TrafficFlow, String, String, TimeWindow> {

    @Override
    public void apply(String s, TimeWindow window, Iterable<TrafficFlow> input, Collector<String> out) throws Exception {
        // 获取窗口开始和结束时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

        // 对窗口中数据进行计算
        Long LicensePlateCount = 0L;
        for (TrafficFlow pageViewBean : input) {
            LicensePlateCount += pageViewBean.getLicensePlateCount();
        }

        // 拼接字符串
        String output = windowStartTime + "," +
                windowEndTime +"," +
                s + "," +
                LicensePlateCount + "," +
                System.currentTimeMillis();

        out.collect(output);
        System.out.println("输出结果 : " + output);
    }

}
